// org/cancer/service/GenderRatioCancerService.scala
package org.cancer.service

import org.apache.spark.sql.{DataFrame, Dataset, SaveMode}
import org.cancer.common.TService
import org.cancer.util.{JDBCUtil, SparkUtil}
import org.apache.spark.sql.functions._

class GenderRatioCancerService_Sylvia extends TService[Dataset[String]] {
  override def dataAnalysis(data: Dataset[String]): Unit = {
    val spark = SparkUtil.takeSpark()
    import spark.implicits._

    // 解析CSV数据并设置表头
    val header = data.first()
    val patientDF = spark.read
      .option("header", "true")
      .csv("input/ChineseCancerPatientInfo.csv")

    // 数据预处理：过滤无效数据
    val cleanedDF = patientDF
      .select("肿瘤类型", "性别")
      .filter($"肿瘤类型".isNotNull && $"性别".isNotNull)
      .filter($"性别" =!= "未知") // 过滤未知性别数据

    // 统计各肿瘤类型的男女人数
    val genderCountDF = cleanedDF
      .groupBy("肿瘤类型", "性别")
      .agg(count("*").alias("人数"))

    // 计算各肿瘤类型的总人数
    val totalCountDF = genderCountDF
      .groupBy("肿瘤类型")
      .agg(sum("人数").alias("总人数"))

    // 计算性别比例
    val genderRatioDF = genderCountDF
      .join(totalCountDF, "肿瘤类型")
      .withColumn("性别比例", round($"人数" / $"总人数", 2))

    // 转换为每个肿瘤类型一行，包含男性和女性比例
    val resultDF = genderRatioDF
      .groupBy("肿瘤类型")
      .pivot("性别")
      .agg(first("性别比例"))
      .select(
        $"肿瘤类型",
        col("男").alias("男性比例"),
        col("女").alias("女性比例")
      )
      .na.fill(0.0) // 填充可能的空值

    // 显示结果
    resultDF.show()

    // 存入数据库
    resultDF.write
      .format("jdbc")
      .option("url", "jdbc:mysql://node1:3306/cancer_patients")
      .option("driver", "com.mysql.cj.jdbc.Driver")
      .option("user", "root")
      .option("password", "123456")
      .option("dbtable", "GenderRatioCancer_Sylvia")
      .mode(SaveMode.Append)
      .save()
  }
}